import { BehaviorSubject, EMPTY, finalize, mergeMap, type Observable, Subject, tap, timer } from "rxjs";

/**
 * 类似生产线方式处理数据
 * 
 * 封装好后，只关注数据的入，和生产线的启动和停止
 */
export class RxjsFlow<T, U = any>
{
    private subject = new Subject<T>()

    /**
     * 定义如何处理数据
     * @param obsMaker 
     */
    constructor(private obsMaker: (sub: Subject<T>) => Observable<U>){}

    isComplete: boolean = false
    private isRun:boolean = false
    private autoCompleteList: RxjsFlow<any>[] = []

    /**
     * 开始运行生产线
     * 
     * 只能被调用一次
     * @param enableAutoComplete 是否启动定时检测其他生产线实现自动停止
     * @returns 
     */
    runObs(enableAutoComplete: boolean = false): Observable<U>
    {
        if(this.isRun) throw new Error("RxjsFlow.runObs: already run")
        this.isRun = true
        if(enableAutoComplete)
        {
            let checkSub = new BehaviorSubject<boolean>(true)
            checkSub.pipe(
                mergeMap(neetCheck => 
                {
                    if(!neetCheck)
                    {
                        this.complete()
                        checkSub.complete()
                        return EMPTY
                    } 
                    return timer(1).pipe(
                        tap(() => checkSub.next(!this.autoCompleteCheck()))
                    )
                }),
            ).subscribe()
        }
        return this.obsMaker(this.subject).pipe(
            finalize(() => this.isComplete = true)
        )
    }

    /**
     * 把其他生产线的运作状态添加到自动停止检测
     * @param ff 
     * @returns 
     */
    addToCompleteCheck(...ff: RxjsFlow<any>[])
    {
        this.autoCompleteList.push(...ff)
        return this
    }

    private autoCompleteCheck()
    {
        let isAllComplete = true
        for(let item of this.autoCompleteList)
        {
            if(item.isComplete == false)
            {
                isAllComplete = false
                break
            }
        }
        return isAllComplete
    }

    /**
     * 往生产线内发送一个数据
     * @param value 
     * @returns 
     */
    next(value: T)
    {
        this.subject.next(value)
        return this
    }

    /**
     * 结束生产线的运行
     * @returns
     */
    complete()
    {
        this.subject.complete()
        return this
    }
}